package com.whty.mongodemo.mongo;

import com.whty.mongodemo.constant.EventBusAddress;
import com.whty.mongodemo.constant.ParamKey;
import com.whty.mongodemo.util.DateUtil;
import io.vertx.core.Future;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.mongo.FindOptions;
import io.vertx.ext.mongo.MongoClient;
import io.vertx.ext.mongo.MongoClientDeleteResult;
import io.vertx.ext.mongo.UpdateOptions;
import io.vertx.ext.mongo.impl.codec.json.JsonObjectCodec;
import lombok.extern.slf4j.Slf4j;
import java.util.List;


@Slf4j
public class MongoListener {

  private final EventBus eventBus;

  private final MongoClient client;

  public MongoListener(EventBus eventBus, MongoClient client) {
    this.eventBus = eventBus;
    this.client = client;
    this.addConsumer();
  }

  private void addConsumer() {
    this.queryById();
    this.query();
    this.insert();
    this.remove();
    this.update();
    this.queryByPage();
    this.queryWithLookup();
  }

  private void queryById() {
    eventBus.localConsumer(EventBusAddress.QUERY_BY_ID, message -> {
      JsonObject body = (JsonObject) message.body();
      String collection = body.getString(ParamKey.COLLECTION);
      JsonObject query = body.getJsonObject(ParamKey.QUERY);
      client.find(collection, query, ar -> {
        if (ar.succeeded()) {
          List<JsonObject> list = ar.result();
          if (list != null && list.size() > 0) {
            JsonObject result = list.get(0);
            message.reply(this.handleDate(result));
          } else {
            message.reply(null);
          }
        } else {
          log.error(ar.cause().getMessage(), ar.cause());
          message.reply(ar.cause());
        }
      });
    });
  }

  private void query() {
    eventBus.localConsumer(EventBusAddress.QUERY, message -> {
      JsonObject body = (JsonObject) message.body();
      String collection = body.getString(ParamKey.COLLECTION);
      JsonObject query = body.getJsonObject(ParamKey.QUERY);
      client.find(collection, query, ar -> {
        if (ar.succeeded()) {
          List<JsonObject> list = ar.result();
          if (list != null && list.size() > 0) {
            list.forEach(this::handleDate);
            message.reply(new JsonArray(list));
          } else {
            message.reply(new JsonArray());
          }
        } else {
          log.error(ar.cause().getMessage(), ar.cause());
          message.reply(ar.cause());
        }
      });
    });
  }

  private void queryByPage() {
    eventBus.localConsumer(EventBusAddress.QUERY_BY_PAGE, message -> {
      JsonObject body = (JsonObject) message.body();
      String collection = body.getString(ParamKey.COLLECTION);
      JsonObject query = body.getJsonObject(ParamKey.QUERY);
      JsonObject pageParams = body.getJsonObject(ParamKey.PAGE);
      Integer pageSIze = pageParams.getInteger(ParamKey.PAGE_SIZE);
      Integer pageNum = pageParams.getInteger(ParamKey.PAGE_NUM);
      JsonObject sort = pageParams.getJsonObject(ParamKey.SORT);
      FindOptions findOptions = new FindOptions()
        .setSort(sort)
        .setLimit(pageSIze)
        .setSkip((pageNum - 1) * pageSIze);
      Future<Long> countFuture = client.count(collection, query);
      Future<List<JsonObject>> dataFuture = client.findWithOptions(collection, query, findOptions);
      Future.all(countFuture, dataFuture).onSuccess(ar -> {
        Long count = ar.resultAt(0);
        List<JsonObject> list = ar.resultAt(1);
        if (list != null && list.size() > 0) {
          list.forEach(this::handleDate);
        }
        JsonObject result = new JsonObject()
          .put(ParamKey.PAGE_SIZE, pageSIze)
          .put(ParamKey.PAGE_NUM, pageNum)
          .put("total", count)
          .put("list", list != null && list.size() > 0 ? list : new JsonArray());
        message.reply(result);
      }).onFailure(e -> {
        log.error(e.getMessage(), e);
        message.reply(e);
      });
    });
  }

  private void insert() {
    eventBus.localConsumer(EventBusAddress.INSERT, message -> {
      JsonObject body = (JsonObject) message.body();
      String collection = body.getString(ParamKey.COLLECTION);
      JsonObject query = body.getJsonObject(ParamKey.QUERY);
      query.put("createTime", new JsonObject().put("$date", DateUtil.getUTCTime()));
      try {
        client.insert(collection, query, ar -> {
          if (ar.succeeded()) {
            String id = ar.result();
            message.reply(new JsonObject().put("_id", id));
          } else {
            log.error("存储数据异常：{}", ar.cause().getMessage(), ar.cause());
            message.reply(ar.cause());
          }
        });
      } catch (Exception e) {
        log.error("存储数据异常：{}", e.getMessage(), e);
        message.reply(e);
      }
    });
  }

  private void update() {
    eventBus.localConsumer(EventBusAddress.UPDATE, message -> {
      JsonObject body = (JsonObject) message.body();
      String collection = body.getString(ParamKey.COLLECTION);
      JsonObject query = body.getJsonObject(ParamKey.QUERY);
      JsonObject data = body.getJsonObject("data");
      data.put("updateTime", new JsonObject().put("$date", DateUtil.getUTCTime()));
      JsonObject update = new JsonObject().put("$set", data);
      //multi set to true to update multiple documents
      client.updateCollectionWithOptions(collection, query, update,
        new UpdateOptions().setMulti(body.containsKey("multi")),
        ar -> {
          if (ar.succeeded()) {
            message.reply(ar.result().toJson());
          } else {
            log.error(ar.cause().getMessage(), ar.cause());
            message.reply(ar.cause());
          }
        });
    });
  }

  private void remove() {
    eventBus.localConsumer(EventBusAddress.REMOVE, message -> {
      JsonObject body = (JsonObject) message.body();
      String collection = body.getString(ParamKey.COLLECTION);
      JsonObject query = body.getJsonObject(ParamKey.QUERY);
      Future<MongoClientDeleteResult> future;
      if (body.containsKey(ParamKey.SINGLE)) { //仅删除匹配的第一条数据
        future = client.removeDocument(collection, query);
      } else {
        future = client.removeDocuments(collection, query);
      }
      future.andThen(ar -> {
        if (ar.succeeded()) {
          message.reply(ar.result().toJson());
        } else {
          log.error(ar.cause().getMessage(), ar.cause());
          message.reply(ar.cause());
        }
      });
    });
  }

  private void queryWithLookup() {
    eventBus.localConsumer(EventBusAddress.QUERY_WITH_LOOKUP, message -> {
      JsonObject body = (JsonObject) message.body();
      String collection = body.getString(ParamKey.COLLECTION);
      JsonObject query = body.getJsonObject(ParamKey.QUERY);
      JsonArray pipeline = new JsonArray();
      JsonObject lookup = new JsonObject()
        .put("$lookup", body.getJsonObject("lookup"));
      pipeline.add(lookup);
      if (null != query) {
        pipeline.add(new JsonObject().put("$match", query));
      }
      JsonArray resultArr = new JsonArray();
      client.aggregate(collection, pipeline)
        .handler(resultArr::add)
        .endHandler(v ->
          message.reply(resultArr)
        ).exceptionHandler(e -> {
          log.error(e.getMessage(), e);
          message.reply(e);
        });
    });

  }

  private JsonObject handleDate(JsonObject json) {
    return this.decodeDateKey(this.decodeDateKey(json, "createTime"), "updateTime");
  }

  private JsonObject decodeDateKey(JsonObject json, String key) {
    Object timeField = json.getValue(key, null);
    if (!(timeField instanceof JsonObject)) return json;
    Object timeString = ((JsonObject) timeField).getValue(JsonObjectCodec.DATE_FIELD, null);
    if (!(timeString instanceof String)) return json;
    json.put(key, DateUtil.formatOffsetDateTime((String) timeString));
    return json;
  }
}
